package concurrent.pool;

import java.util.HashSet;
import java.util.Set;
import java.util.concurrent.TimeUnit;

/**
 * @author Jay
 */
public class ThreadPool {
    private final BlockingQueue<Runnable> taskQueue;
    private final Set<Worker> workers;
    private final int corePoolSize;
    private final long timeout;
    private final TimeUnit timeUnit;
    private final RejectPolicy<Runnable> rejectPolicy;

    static final long DEFAULT_TIMEOUT = 0L;
    static final RejectPolicy<Runnable> ABORT_POLICY = new AbortPolicy();
    static final RejectPolicy<Runnable> DISCARD_POLICY = new DiscardPolicy();
    static final RejectPolicy<Runnable> DISCARD_OLDEST_POLICY = new DiscardOldestPolicy();
    static final RejectPolicy<Runnable> CALLER_RUN_POLICY = new CallerRunsPolicy();

    public ThreadPool(int corePoolSize, int taskQueueCapacity, long timeout, TimeUnit timeUnit, RejectPolicy<Runnable> rejectPolicy) {
        if(corePoolSize <= 0 || taskQueueCapacity <= 0 || timeout < 0) {
            throw new IllegalArgumentException();
        }
        this.corePoolSize = corePoolSize;
        this.timeout = timeout;
        this.timeUnit = timeUnit;
        this.taskQueue = new BlockingQueue<>(taskQueueCapacity);
        this.workers = new HashSet<>(corePoolSize);
        this.rejectPolicy = rejectPolicy;
    }

    public ThreadPool(int corePoolSize, int taskQueueCapacity, long timeout, TimeUnit timeUnit) {
        if(corePoolSize <= 0 || taskQueueCapacity <= 0 || timeout < 0) {
            throw new IllegalArgumentException();
        }
        this.corePoolSize = corePoolSize;
        this.timeout = timeout;
        this.timeUnit = timeUnit;
        this.taskQueue = new BlockingQueue<>(taskQueueCapacity);
        this.workers = new HashSet<>(corePoolSize);
        this.rejectPolicy = ABORT_POLICY;
    }

    /**
     * 创建大小固定的线程池，使用默认保活时间和拒绝策略
     * @param corePoolSize 线程池大小
     * @param taskQueueCapacity 任务队列大小
     * @return 固定大小的线程池
     */
    public static ThreadPool newFixedThreadPool(int corePoolSize, int taskQueueCapacity){
        return new ThreadPool(corePoolSize, taskQueueCapacity, DEFAULT_TIMEOUT, TimeUnit.MILLISECONDS);
    }

    public void execute(Runnable task){
        synchronized (workers){
            // 工作线程集合未满，创建新worker
            if(workers.size() < corePoolSize){
                Worker worker = new Worker(task);
                workers.add(worker);
                worker.start();
            }
            else{
                // 工作线程集合已满，尝试将任务放入任务队列
                taskQueue.tryPut(task, rejectPolicy);
            }
        }
    }


    class Worker extends Thread{
        private Runnable task;

        public Worker(Runnable task) {
            this.task = task;
        }

        @Override
        public void run() {
            // 先执行已获得的任务，再向队列申请新任务（如果超时仍未获得新任务，线程死亡）
            while(task != null || (task = taskQueue.poll(timeout, timeUnit)) != null){
                try{
                    task.run();
                }
                catch (Exception e){
                    e.printStackTrace();
                }
                finally {
                    task = null;
                }
            }
        }

        @Override
        public int hashCode() {
            return super.hashCode();
        }

        @Override
        public boolean equals(Object obj) {
            return super.equals(obj);
        }
    }

    /**
     * 中止策略，不接受任务，直接抛出异常
     */
    static class AbortPolicy implements RejectPolicy<Runnable>{

        @Override
        public void reject(BlockingQueue<Runnable> queue, Runnable task) {
            throw new RuntimeException("无法接受更多任务");
        }
    }

    /**
     * 调用者执行策略，将任务交给调用线程执行
     */
    static class CallerRunsPolicy implements RejectPolicy<Runnable>{
        @Override
        public void reject(BlockingQueue<Runnable> queue, Runnable task) {
            task.run();
        }
    }

    /**
     * 放弃策略，放弃本次任务
     */
    static class DiscardPolicy implements RejectPolicy<Runnable>{
        @Override
        public void reject(BlockingQueue<Runnable> queue, Runnable task) {

        }
    }

    /**
     * 放弃最早任务策略，放弃队列中最早的任务，用本次任务替换
     */
    static class DiscardOldestPolicy implements RejectPolicy<Runnable>{

        @Override
        public void reject(BlockingQueue<Runnable> queue, Runnable task) {
            queue.lock();
            try{
                queue.removeFirst();
                queue.addLast(task);
            }
            finally {
                queue.unlock();
            }
        }
    }
}
